package com.iwhale.sequence.redisidx.sequence;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.iwhale.sequence.redisidx.config.SequenceConfig;
import com.iwhale.sequence.redisidx.utils.ApplicationContextUtil;

import io.netty.util.concurrent.DefaultThreadFactory;

public class SequenceImpl implements Sequence {
    // logger
    private static Logger logger = LoggerFactory.getLogger(SequenceImpl.class);

    // 序列默认18位
//    private static final Long START_SEQ = 100000000000000000L;

    private final ReentrantLock lock = new ReentrantLock(true);

    private final Condition notEmpty = lock.newCondition();

    private final AtomicBoolean increasing = new AtomicBoolean(false);

    private SequenceRedis sequenceRedis;

    private String seqName;

    // 配置信息
    private SequenceConfig sequenceConfig;

    // 无须闲置长度，同一时刻，同一序列只会提交一个任务，即最大数量和序列数量一致。
    private static final BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();

    private static final ExecutorService pool = new ThreadPoolExecutor(4, 16, 60L, TimeUnit.MILLISECONDS, workQueue,
        new DefaultThreadFactory("seq-incr"));

    public SequenceImpl(String seqName) {
        sequenceRedis = ApplicationContextUtil.getBean(SequenceRedis.class);
        sequenceConfig = ApplicationContextUtil.getBean(SequenceConfig.class);
        this.seqName = seqName;
        logger.info("新建序列：{}", seqName);
    }

    // 只考虑序列服务器单节点，即认为从redis获取段是连贯的，不存在被其他节点取走。
    private final AtomicLong sequence = new AtomicLong(0);

    private final AtomicLong maxSequence = new AtomicLong(0);

    public Long nextVal() throws Exception {
        final int timeOut = 1;
        final ReentrantLock lock = this.lock;

        // 起始seq
        Long startSeq = sequenceConfig.getStartSequence();
        // 获取下一步序列阈值
        int threshold = sequenceConfig.getPaddingThreshold();

        lock.lockInterruptibly();
        long nanos = TimeUnit.SECONDS.toNanos(timeOut);
        try {
            // 如果序列已经用尽了，就等着从redis里面获取
            while (sequence.get() == maxSequence.get()) {
                if (nanos <= 0) {
                    // 等1秒钟，1秒钟都还没有OK，直接异常. 性能问题或者线程死锁
                    throw new Exception("等待超时：" + timeOut + "秒！");
                }
                // 提交任务，获取最新的序列
                addIncrTask();
                // 等着从redis获取最新的序列，最多等待1ms
                nanos = notEmpty.awaitNanos(nanos);
            }
            long seq = sequence.incrementAndGet();
            // 使用到50%时，再次获取序列
            if (maxSequence.get() - seq < threshold) {
                addIncrTask();
            }
            if (seq < maxSequence.get()) {
                notEmpty.signal();
            }
            return seq + startSeq;
        }
        finally {
            lock.unlock();
        }
    }

    private void addIncrTask() {
        // 多线程并发时只调用一次
        if (!increasing.getAndSet(true)) {
            pool.submit(incrTask);
        }
    }

    private final Runnable incrTask = () -> {
        final ReentrantLock lock = this.lock;
        final int delta = this.sequenceConfig.getDelta();
        try {
            long newIdx = sequenceRedis.increment(seqName, delta);
            lock.lock();
            try {
                // 启动之后，第一次初始化
                if (sequence.get() == 0) {
                    sequence.set(newIdx - delta);
                }
                maxSequence.set(newIdx);
                increasing.set(false);
                notEmpty.signal();
            }
            finally {
                lock.unlock();
            }
        }
        catch (Exception e) {
            logger.error("redis取序列：" + seqName + "失败。", e);
        }
        finally {
            increasing.set(false);
        }
    };

}
